/*-------------------------------------------------------------------------
 *
 * execReplication.c
 *	  逻辑复制的各种执行例程
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/executor/execReplication.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/genam.h"
#include "access/relscan.h"
#include "access/tableam.h"
#include "access/transam.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_relation.h"
#include "parser/parsetree.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/typcache.h"


/*
 * 为关系 'rel' 中的元组 'key' 设置 ScanKey，以匹配 'rel'
 * （*不是* idxrel！）。
 *
 * 返回是否任何列包含 NULL。
 *
 * 这不是通用例程，它期望 idxrel 是 rel 的复制身份并符合与之
 * 相关的所有限制。
 */
static bool fc_build_replindex_scan_key(ScanKey fc_skey, Relation fc_rel, Relation fc_idxrel,
						 TupleTableSlot *fc_searchslot)
{
	int			fc_attoff;
	bool		fc_isnull;
	Datum		fc_indclassDatum;
	oidvector  *fc_opclass;
	int2vector *fc_indkey = &fc_idxrel->rd_index->indkey;
	bool		fc_hasnulls = false;

	Assert(RelationGetReplicaIndex(fc_rel) == RelationGetRelid(fc_idxrel) ||
		   RelationGetPrimaryKeyIndex(fc_rel) == RelationGetRelid(fc_idxrel));

	fc_indclassDatum = SysCacheGetAttr(INDEXRELID, fc_idxrel->rd_indextuple,
									Anum_pg_index_indclass, &fc_isnull);
	Assert(!fc_isnull);
	fc_opclass = (oidvector *) DatumGetPointer(fc_indclassDatum);

	/* 为索引中的每个属性构建 scankey。 */
	for (fc_attoff = 0; fc_attoff < IndexRelationGetNumberOfKeyAttributes(fc_idxrel); fc_attoff++)
	{
		Oid			fc_operator;
		Oid			fc_opfamily;
		RegProcedure fc_regop;
		int			fc_pkattno = fc_attoff + 1;
		int			fc_mainattno = fc_indkey->values[fc_attoff];
		Oid			fc_optype = get_opclass_input_type(fc_opclass->values[fc_attoff]);

		/*
		 * 加载操作符信息。我们需要此信息来获取扫描键的
		 * 相等操作符函数。
		 */
		fc_opfamily = get_opclass_family(fc_opclass->values[fc_attoff]);

		fc_operator = get_opfamily_member(fc_opfamily, fc_optype,
									   fc_optype,
									   BTEqualStrategyNumber);
		if (!OidIsValid(fc_operator))
			elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
				 BTEqualStrategyNumber, fc_optype, fc_optype, fc_opfamily);

		fc_regop = get_opcode(fc_operator);

		/* 初始化 scankey。 */
		ScanKeyInit(&fc_skey[fc_attoff],
					fc_pkattno,
					BTEqualStrategyNumber,
					fc_regop,
					fc_searchslot->tts_values[fc_mainattno - 1]);

		fc_skey[fc_attoff].sk_collation = fc_idxrel->rd_indcollation[fc_attoff];

		/* 检查空值。 */
		if (fc_searchslot->tts_isnull[fc_mainattno - 1])
		{
			fc_hasnulls = true;
			fc_skey[fc_attoff].sk_flags |= SK_ISNULL;
		}
	}

	return fc_hasnulls;
}

/*
 * 使用索引在关系 'rel' 中搜索元组。
 *
 * 如果找到匹配的元组，则用 lockmode 锁定它，用其
 * 内容填充槽，并返回 true。否则返回 false。
 */
bool RelationFindReplTupleByIndex(Relation fc_rel, Oid fc_idxoid,
							 LockTupleMode fc_lockmode,
							 TupleTableSlot *fc_searchslot,
							 TupleTableSlot *fc_outslot)
{
	ScanKeyData fc_skey[INDEX_MAX_KEYS];
	IndexScanDesc fc_scan;
	SnapshotData fc_snap;
	TransactionId fc_xwait;
	Relation	fc_idxrel;
	bool		fc_found;

	/* 打开索引。 */
	fc_idxrel = index_open(fc_idxoid, RowExclusiveLock);

	/* 开始索引扫描。 */
	InitDirtySnapshot(fc_snap);
	fc_scan = index_beginscan(fc_rel, fc_idxrel, &fc_snap,
						   IndexRelationGetNumberOfKeyAttributes(fc_idxrel),
						   0);

	/* 构建扫描键。 */
	fc_build_replindex_scan_key(fc_skey, fc_rel, fc_idxrel, fc_searchslot);

retry:
	fc_found = false;

	index_rescan(fc_scan, fc_skey, IndexRelationGetNumberOfKeyAttributes(fc_idxrel), NULL, 0);

	/* 尝试找到元组 */
	if (index_getnext_slot(fc_scan, ForwardScanDirection, fc_outslot))
	{
		fc_found = true;
		ExecMaterializeSlot(fc_outslot);

		fc_xwait = TransactionIdIsValid(fc_snap.xmin) ?
			fc_snap.xmin : fc_snap.xmax;

		/*
		 * 如果元组被锁定，等待锁定事务完成并
		 * 重试。
		 */
		if (TransactionIdIsValid(fc_xwait))
		{
			XactLockTableWait(fc_xwait, NULL, NULL, XLTW_None);
			goto retry;
		}
	}

	/* 找到元组，尝试以锁定模式锁定它。 */
	if (fc_found)
	{
		TM_FailureData fc_tmfd;
		TM_Result	fc_res;

		PushActiveSnapshot(GetLatestSnapshot());

		fc_res = table_tuple_lock(fc_rel, &(fc_outslot->tts_tid), GetLatestSnapshot(),
							   fc_outslot,
							   GetCurrentCommandId(false),
							   fc_lockmode,
							   LockWaitBlock,
							   0 /* 不跟随更新 */ ,
							   &fc_tmfd);

		PopActiveSnapshot();

		switch (fc_res)
		{
			case TM_Ok:
				break;
			case TM_Updated:
				/* XXX: 改进此处的处理 */
				if (ItemPointerIndicatesMovedPartitions(&fc_tmfd.ctid))
					ereport(LOG,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
				else
					ereport(LOG,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("concurrent update, retrying")));
				goto retry;
			case TM_Deleted:
				/* XXX: 改进此处的处理 */
				ereport(LOG,
						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
						 errmsg("concurrent delete, retrying")));
				goto retry;
			case TM_Invisible:
				elog(ERROR, "attempted to lock invisible tuple");
				break;
			default:
				elog(ERROR, "unexpected table_tuple_lock status: %u", fc_res);
				break;
		}
	}

	index_endscan(fc_scan);

	/* 不要在提交之前释放锁。 */
	index_close(fc_idxrel, NoLock);

	return fc_found;
}

/*
 * 通过检查槽中的元组值是否相等来比较元组。
 */
static bool fc_tuples_equal(TupleTableSlot *fc_slot1, TupleTableSlot *fc_slot2,
			 TypeCacheEntry **fc_eq)
{
	int			fc_attrnum;

	Assert(fc_slot1->tts_tupleDescriptor->natts ==
		   fc_slot2->tts_tupleDescriptor->natts);

	slot_getallattrs(fc_slot1);
	slot_getallattrs(fc_slot2);

	/* 检查属性的相等性。 */
	for (fc_attrnum = 0; fc_attrnum < fc_slot1->tts_tupleDescriptor->natts; fc_attrnum++)
	{
		Form_pg_attribute fc_att;
		TypeCacheEntry *fc_typentry;

		fc_att = TupleDescAttr(fc_slot1->tts_tupleDescriptor, fc_attrnum);

		/*
		 * 忽略已删除和生成的列，因为发布者不会发送这些
		 */
		if (fc_att->attisdropped || fc_att->attgenerated)
			continue;

		/*
		 * 如果一个值为NULL而另一个值不是，则它们肯定不相等
		 */
		if (fc_slot1->tts_isnull[fc_attrnum] != fc_slot2->tts_isnull[fc_attrnum])
			return false;

		/*
		 * 如果两个都是NULL，则可以认为它们是相等的。
		 */
		if (fc_slot1->tts_isnull[fc_attrnum] || fc_slot2->tts_isnull[fc_attrnum])
			continue;

		fc_typentry = fc_eq[fc_attrnum];
		if (fc_typentry == NULL)
		{
			fc_typentry = lookup_type_cache(fc_att->atttypid,
										 TYPECACHE_EQ_OPR_FINFO);
			if (!OidIsValid(fc_typentry->eq_opr_finfo.fn_oid))
				ereport(ERROR,
						(errcode(ERRCODE_UNDEFINED_FUNCTION),
						 errmsg("could not identify an equality operator for type %s",
								format_type_be(fc_att->atttypid))));
			fc_eq[fc_attrnum] = fc_typentry;
		}

		if (!DatumGetBool(FunctionCall2Coll(&fc_typentry->eq_opr_finfo,
											fc_att->attcollation,
											fc_slot1->tts_values[fc_attrnum],
											fc_slot2->tts_values[fc_attrnum])))
			return false;
	}

	return true;
}
#ifdef FDDM
//根据索引包含的属性对应的值，判断两个记录是否相等
static bool fdd_tuples_equal_indkey(TupleTableSlot *fc_slot1, TupleTableSlot *fc_slot2,
			 TypeCacheEntry **fc_eq, int *fc_indkeyAttrNos, int fc_keyAttrNum)
{
	int fc_i = 0;

	Assert(fc_slot1->tts_tupleDescriptor->natts ==
		   fc_slot2->tts_tupleDescriptor->natts);

	slot_getallattrs(fc_slot1);
	slot_getallattrs(fc_slot2);

	if (fc_keyAttrNum > fc_slot1->tts_tupleDescriptor->natts)
	{
		return false;
	}

	/* 检查属性的相等性。 */
	for (fc_i = 0; fc_i < fc_keyAttrNum; fc_i++)
	{
		Form_pg_attribute fc_att;
		TypeCacheEntry *fc_typentry;
		int fc_attrnum = fc_indkeyAttrNos[fc_i];

		/*
		 * 如果一个值为NULL而另一个值不是，则它们肯定不相等
		 */
		if (fc_slot1->tts_isnull[fc_attrnum] != fc_slot2->tts_isnull[fc_attrnum])
			return false;

		/*
		 * 如果两个都是NULL，则可以认为它们是相等的。
		 */
		if (fc_slot1->tts_isnull[fc_attrnum] || fc_slot2->tts_isnull[fc_attrnum])
			continue;

		fc_att = TupleDescAttr(fc_slot1->tts_tupleDescriptor, fc_attrnum);

		fc_typentry = fc_eq[fc_attrnum];
		if (fc_typentry == NULL)
		{
			fc_typentry = lookup_type_cache(fc_att->atttypid,
										 TYPECACHE_EQ_OPR_FINFO);
			if (!OidIsValid(fc_typentry->eq_opr_finfo.fn_oid))
				ereport(ERROR,
						(errcode(ERRCODE_UNDEFINED_FUNCTION),
						 errmsg("could not identify an equality operator for type %s",
								format_type_be(fc_att->atttypid))));
			fc_eq[fc_attrnum] = fc_typentry;
		}

		if (!DatumGetBool(FunctionCall2Coll(&fc_typentry->eq_opr_finfo,
											fc_att->attcollation,
											fc_slot1->tts_values[fc_attrnum],
											fc_slot2->tts_values[fc_attrnum])))
			return false;
	}

	return true;
}

//查找指定xmin的可见行
bool FDD_RelationFindReplTupleSeqByXmin(Relation fc_rel, LockTupleMode fc_lockmode, TupleTableSlot *fc_outslot, 	TransactionId fc_xmin)
{
	TupleTableSlot *fc_scanslot;
	TableScanDesc fc_scan;
	SnapshotData fc_snap;
	TypeCacheEntry **fc_eq;
	TransactionId fc_xwait;
	bool		fc_found;
	TupleDesc	fc_desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(fc_rel);



	Assert(equalTupleDescs(fc_desc, fc_outslot->tts_tupleDescriptor));

	fc_eq = palloc0(sizeof(*fc_eq) * fc_outslot->tts_tupleDescriptor->natts);

	/* 开始堆扫描。 */
	InitDirtySnapshot(fc_snap);
	//可见行
	fc_scan = table_beginscan(fc_rel, &fc_snap, 0, NULL);
	fc_scanslot = table_slot_create(fc_rel, NULL);

retry:
	fc_found = false;

	table_rescan(fc_scan, NULL);

	/* 尝试找到元组 */
	while (table_scan_getnextslot(fc_scan, ForwardScanDirection, fc_scanslot))
	{
		HeapTuple fc_tuple = (((HeapTupleTableSlot *)fc_scanslot)->tuple);
		TransactionId fc_tp_min = HeapTupleHeaderGetRawXmin(fc_tuple->t_data);

		if (fc_tp_min != fc_xmin)
		{
			continue;
		}
			
		fc_found = true;
		ExecCopySlot(fc_outslot, fc_scanslot);

		fc_xwait = TransactionIdIsValid(fc_snap.xmin) ?
			fc_snap.xmin : fc_snap.xmax;

		/*
		 * 如果元组被锁定，等待锁定事务完成并
		 * 重试。
		 */
		if (TransactionIdIsValid(fc_xwait))
		{
			XactLockTableWait(fc_xwait, NULL, NULL, XLTW_None);
			goto retry;
		}

		/* 找到了我们的元组，它没有被锁定 */
		break;
	}

	/* 找到元组，尝试以锁定模式锁定它。 */
	if (fc_found)
	{
		TM_FailureData fc_tmfd;
		TM_Result	fc_res;

		PushActiveSnapshot(GetLatestSnapshot());

		fc_res = table_tuple_lock(fc_rel, &(fc_outslot->tts_tid), GetLatestSnapshot(),
							   fc_outslot,
							   GetCurrentCommandId(false),
							   fc_lockmode,
							   LockWaitBlock,
							   0 /* 不跟随更新 */ ,
							   &fc_tmfd);

		PopActiveSnapshot();

		switch (fc_res)
		{
			case TM_Ok:
				break;
			case TM_Updated:
				/* XXX: 改进此处的处理 */
				if (ItemPointerIndicatesMovedPartitions(&fc_tmfd.ctid))
					ereport(LOG,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
				else
					ereport(LOG,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("concurrent update, retrying")));
				goto retry;
			case TM_Deleted:
				/* XXX: 改进此处的处理 */
				ereport(LOG,
						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
						 errmsg("concurrent delete, retrying")));
				goto retry;
			case TM_Invisible:
				elog(ERROR, "attempted to lock invisible tuple");
				break;
			default:
				elog(ERROR, "unexpected table_tuple_lock status: %u", fc_res);
				break;
		}
	}

	table_endscan(fc_scan);
	ExecDropSingleTupleTableSlot(fc_scanslot);

	return fc_found;
}

//RelationFindReplTupleSeq查找死行版，因为里面有很多静态函数，不好移到插件代码中
//dead_tup 是否查找死行，查找死行也进行全表扫描，找xmax有值的，并切xmax是最大的
//indkeyAttrNos 根据某个索引查找死行，这里保存的是这个索引包含的属性的序号
//keyAttrNum    索引包含属性的个数
//1、用来查找死行，2、用来查找indenty full 可见行
bool FDD_RelationFindReplTupleSeqDead(Relation fc_rel, LockTupleMode fc_lockmode, TupleTableSlot *fc_searchslot, TupleTableSlot *fc_outslot,
	bool fc_dead_tup, int *fc_indkeyAttrNos, int fc_keyAttrNum, int *fc_retVal)
{
	TupleTableSlot *fc_scanslot;
	TupleTableSlot *fc_tmp_slot = NULL;
	TableScanDesc fc_scan;
	SnapshotData fc_snap;
	TypeCacheEntry **fc_eq;
	TransactionId fc_xwait;
	bool		fc_found;
	TransactionId fc_dead_line_xmax = InvalidTransactionId;
	TupleDesc	fc_desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(fc_rel);

	Assert(equalTupleDescs(fc_desc, fc_outslot->tts_tupleDescriptor));

	fc_eq = palloc0(sizeof(*fc_eq) * fc_outslot->tts_tupleDescriptor->natts);

	/* 开始堆扫描。 */
	if (!fc_dead_tup)
	{
		InitDirtySnapshot(fc_snap);
	}
	else
	{
		fc_snap.snapshot_type = SNAPSHOT_ANY;
	}
	
	fc_scan = table_beginscan(fc_rel, &fc_snap, 0, NULL);
	fc_scanslot = table_slot_create(fc_rel, NULL);

retry:
	fc_found = false;

	table_rescan(fc_scan, NULL);

	/* 尝试找到元组 */
	while (table_scan_getnextslot(fc_scan, ForwardScanDirection, fc_scanslot))
	{
		if (!fc_dead_tup
			|| fc_keyAttrNum <= 0)
		{
			//查找可见元组或完整身份元组
			if (!fc_tuples_equal(fc_scanslot, fc_searchslot, fc_eq))
				continue;
		}
		else
		{
			//死元组，但不是完整身份
			if (!fdd_tuples_equal_indkey(fc_scanslot, fc_searchslot, fc_eq, fc_indkeyAttrNos, fc_keyAttrNum))
				continue;
		}
		
		if (!fc_dead_tup)
		{
			//visible tup
			fc_found = true;
			ExecCopySlot(fc_outslot, fc_scanslot);
			fc_xwait = TransactionIdIsValid(fc_snap.xmin) ?
			fc_snap.xmin : fc_snap.xmax;
			/*
			 * 如果元组被锁定，等待锁定事务完成并重试。
			 */
			if (TransactionIdIsValid(fc_xwait))
			{
				XactLockTableWait(fc_xwait, NULL, NULL, XLTW_None);
				goto retry;
			}

			/* 找到了我们的元组，它没有被锁定 */
			break;
		}
		else
		{
			//可能有多个死元组
			//save
			HeapTuple fc_tuple = (((HeapTupleTableSlot *)fc_scanslot)->tuple);
			HeapTupleHeader fc_tuple_header = fc_tuple->t_data;
			TransactionId fc_tp_max = HeapTupleHeaderGetRawXmax(fc_tuple->t_data);

			if (!(fc_tuple_header->t_infomask & HEAP_XMAX_COMMITTED)
				|| !TransactionIdIsNormal(fc_tp_max)
				|| fc_tp_max == GetCurrentTransactionId())
			{
				//xmax是0，或者没有提交，或者是当前的事务id
				continue;
			}

			
			if (fc_dead_line_xmax == InvalidTransactionId
				|| TransactionIdPrecedes(fc_dead_line_xmax , fc_tp_max))
			{
				if (fc_tmp_slot == NULL)
				{
					fc_tmp_slot = table_slot_create(fc_rel, NULL);
				}
				fc_found = true;
				
				ExecCopySlot(fc_tmp_slot, fc_scanslot);
				fc_dead_line_xmax = fc_tp_max;
			}
		}
	}

	/* 找到元组，尝试以锁定模式锁定它。 */
	if (fc_found)
	{
		TM_FailureData fc_tmfd;
		TM_Result	fc_res;
		TupleTableSlot *fc_to_lock_slot = fc_dead_tup ? fc_tmp_slot : fc_outslot;
		
		PushActiveSnapshot(GetLatestSnapshot());

		fc_res = table_tuple_lock(fc_rel, &(fc_to_lock_slot->tts_tid), GetLatestSnapshot(),
							   fc_to_lock_slot,
							   GetCurrentCommandId(false),
							   fc_lockmode,
							   LockWaitBlock,
							   0 /* 不跟随更新 */ ,
							   &fc_tmfd);

		PopActiveSnapshot();

		switch (fc_res)
		{
			case TM_Ok:
			{
				if (fc_dead_tup)
				{
					//要找死行，找到最近xmax，但是状态是ok，可能被rollback
					fc_found = false;
				}
				break;
			}
			case TM_Updated:
				if (fc_dead_tup)
				{
					//将本地槽复制到输出槽
					ExecCopySlot(fc_outslot, fc_tmp_slot);
					ExecDropSingleTupleTableSlot(fc_tmp_slot);
				}
				/* XXX: 改进此处的处理 */
				if (ItemPointerIndicatesMovedPartitions(&fc_tmfd.ctid))
					ereport(DEBUG1,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
				else
					ereport(DEBUG1,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("concurrent update, retrying")));
				break;
			case TM_Deleted:
				{
				/* XXX: 改进此处的处理 */
					if (fc_dead_tup)
					{
						//将本地槽复制到输出槽
						ExecCopySlot(fc_outslot, fc_tmp_slot);
						ExecDropSingleTupleTableSlot(fc_tmp_slot);
					}
					ereport(DEBUG1,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("concurrent delete")));
				}
				break;
			default:
				elog(ERROR, "unexpected table_tuple_lock status: %u", fc_res);
				break;
		}

		*fc_retVal = (int)fc_res;
	}

	table_endscan(fc_scan);
	ExecDropSingleTupleTableSlot(fc_scanslot);

	return fc_found;
}

#endif
/*
 * 使用顺序扫描在关系 'rel' 中搜索元组。
 *
 * 如果找到匹配的元组，用锁定模式锁定它，填充槽的内容并返回true。否则返回false。
 *
 * 请注意，这在第一个匹配的元组上停止。
 *
 * 显然，在有超过几行的表中，这可能会非常慢。
 */
bool RelationFindReplTupleSeq(Relation fc_rel, LockTupleMode fc_lockmode,
						 TupleTableSlot *fc_searchslot, TupleTableSlot *fc_outslot)
{
	TupleTableSlot *fc_scanslot;
	TableScanDesc fc_scan;
	SnapshotData fc_snap;
	TypeCacheEntry **fc_eq;
	TransactionId fc_xwait;
	bool		fc_found;
	TupleDesc	fc_desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(fc_rel);

	Assert(equalTupleDescs(fc_desc, fc_outslot->tts_tupleDescriptor));

	fc_eq = palloc0(sizeof(*fc_eq) * fc_outslot->tts_tupleDescriptor->natts);

	/* 开始堆扫描。 */
	InitDirtySnapshot(fc_snap);
	fc_scan = table_beginscan(fc_rel, &fc_snap, 0, NULL);
	fc_scanslot = table_slot_create(fc_rel, NULL);

retry:
	fc_found = false;

	table_rescan(fc_scan, NULL);

	/* 尝试找到元组 */
	while (table_scan_getnextslot(fc_scan, ForwardScanDirection, fc_scanslot))
	{
		if (!fc_tuples_equal(fc_scanslot, fc_searchslot, fc_eq))
			continue;

		fc_found = true;
		ExecCopySlot(fc_outslot, fc_scanslot);

		fc_xwait = TransactionIdIsValid(fc_snap.xmin) ?
			fc_snap.xmin : fc_snap.xmax;

		/*
		 * 如果元组被锁定，等待锁定事务完成并
		 * 重试。
		 */
		if (TransactionIdIsValid(fc_xwait))
		{
			XactLockTableWait(fc_xwait, NULL, NULL, XLTW_None);
			goto retry;
		}

		/* 找到了我们的元组，它没有被锁定 */
		break;
	}

	/* 找到元组，尝试以锁定模式锁定它。 */
	if (fc_found)
	{
		TM_FailureData fc_tmfd;
		TM_Result	fc_res;

		PushActiveSnapshot(GetLatestSnapshot());

		fc_res = table_tuple_lock(fc_rel, &(fc_outslot->tts_tid), GetLatestSnapshot(),
							   fc_outslot,
							   GetCurrentCommandId(false),
							   fc_lockmode,
							   LockWaitBlock,
							   0 /* 不跟随更新 */ ,
							   &fc_tmfd);

		PopActiveSnapshot();

		switch (fc_res)
		{
			case TM_Ok:
				break;
			case TM_Updated:
				/* XXX: 改进此处的处理 */
				if (ItemPointerIndicatesMovedPartitions(&fc_tmfd.ctid))
					ereport(LOG,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
				else
					ereport(LOG,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("concurrent update, retrying")));
				goto retry;
			case TM_Deleted:
				/* XXX: 改进此处的处理 */
				ereport(LOG,
						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
						 errmsg("concurrent delete, retrying")));
				goto retry;
			case TM_Invisible:
				elog(ERROR, "attempted to lock invisible tuple");
				break;
			default:
				elog(ERROR, "unexpected table_tuple_lock status: %u", fc_res);
				break;
		}
	}

	table_endscan(fc_scan);
	ExecDropSingleTupleTableSlot(fc_scanslot);

	return fc_found;
}

/*
 * 将槽中表示的元组插入到关系中，更新索引，
 * 并执行任何约束和逐行触发器。
 *
 * 调用者有责任打开索引。
 */
void ExecSimpleRelationInsert(ResultRelInfo *fc_resultRelInfo,
						 EState *fc_estate, TupleTableSlot *fc_slot)
{
	bool		fc_skip_tuple = false;
	Relation	fc_rel = fc_resultRelInfo->ri_RelationDesc;

	/* 目前我们只支持表。 */
	Assert(fc_rel->rd_rel->relkind == RELKIND_RELATION);

	CheckCmdReplicaIdentity(fc_rel, CMD_INSERT);

	/* BEFORE ROW INSERT 触发器 */
	if (fc_resultRelInfo->ri_TrigDesc &&
		fc_resultRelInfo->ri_TrigDesc->trig_insert_before_row)
	{
		if (!ExecBRInsertTriggers(fc_estate, fc_resultRelInfo, fc_slot))
			fc_skip_tuple = true;	/* "不做任何操作" */
	}

	if (!fc_skip_tuple)
	{
		List	   *fc_recheckIndexes = NIL;

		/* 计算存储生成的列 */
		if (fc_rel->rd_att->constr &&
			fc_rel->rd_att->constr->has_generated_stored)
			ExecComputeStoredGenerated(fc_resultRelInfo, fc_estate, fc_slot,
									   CMD_INSERT);

		/* 检查元组的约束 */
		if (fc_rel->rd_att->constr)
			ExecConstraints(fc_resultRelInfo, fc_slot, fc_estate);
		if (fc_rel->rd_rel->relispartition)
			ExecPartitionCheck(fc_resultRelInfo, fc_slot, fc_estate, true);

		/* 好的，存储元组并为其创建索引条目 */
		simple_table_tuple_insert(fc_resultRelInfo->ri_RelationDesc, fc_slot);

		if (fc_resultRelInfo->ri_NumIndices > 0)
			fc_recheckIndexes = ExecInsertIndexTuples(fc_resultRelInfo,
												   fc_slot, fc_estate, false, false,
												   NULL, NIL);

		/* AFTER ROW INSERT 触发器 */
		ExecARInsertTriggers(fc_estate, fc_resultRelInfo, fc_slot,
							 fc_recheckIndexes, NULL);

		/*
		 * XXX 理论上我们应该将TransitionCaptureState对象传递给
		 * 以上内容以捕获过渡元组，但实际上
		 * 语句触发器尚未通过复制触发
		 */

		list_free(fc_recheckIndexes);
	}
}

/*
 * 找到搜索槽元组并用槽中的数据更新它，
 * 更新索引，并执行任何约束和逐行触发器。
 *
 * 调用者有责任打开索引。
 */
void ExecSimpleRelationUpdate(ResultRelInfo *fc_resultRelInfo,
						 EState *fc_estate, EPQState *fc_epqstate,
						 TupleTableSlot *fc_searchslot, TupleTableSlot *fc_slot)
{
	bool		fc_skip_tuple = false;
	Relation	fc_rel = fc_resultRelInfo->ri_RelationDesc;
	ItemPointer fc_tid = &(fc_searchslot->tts_tid);

	/*
	 * 我们只支持非系统表，
	 * check_publication_add_relation()负责。
	 */
	Assert(fc_rel->rd_rel->relkind == RELKIND_RELATION);
	Assert(!IsCatalogRelation(fc_rel));

	CheckCmdReplicaIdentity(fc_rel, CMD_UPDATE);

	/* 行更新之前触发器 */
	if (fc_resultRelInfo->ri_TrigDesc &&
		fc_resultRelInfo->ri_TrigDesc->trig_update_before_row)
	{
		if (!ExecBRUpdateTriggers(fc_estate, fc_epqstate, fc_resultRelInfo,
								  fc_tid, NULL, fc_slot, NULL))
			fc_skip_tuple = true;	/* "不做任何操作" */
	}

	if (!fc_skip_tuple)
	{
		List	   *fc_recheckIndexes = NIL;
		bool		update_indexes;

		/* 计算存储生成的列 */
		if (fc_rel->rd_att->constr &&
			fc_rel->rd_att->constr->has_generated_stored)
			ExecComputeStoredGenerated(fc_resultRelInfo, fc_estate, fc_slot,
									   CMD_UPDATE);

		/* 检查元组的约束 */
		if (fc_rel->rd_att->constr)
			ExecConstraints(fc_resultRelInfo, fc_slot, fc_estate);
		if (fc_rel->rd_rel->relispartition)
			ExecPartitionCheck(fc_resultRelInfo, fc_slot, fc_estate, true);

		simple_table_tuple_update(fc_rel, fc_tid, fc_slot, fc_estate->es_snapshot,
								  &update_indexes);

		if (fc_resultRelInfo->ri_NumIndices > 0 && update_indexes)
			fc_recheckIndexes = ExecInsertIndexTuples(fc_resultRelInfo,
												   fc_slot, fc_estate, true, false,
												   NULL, NIL);

		/* 行更新之后触发器 */
		ExecARUpdateTriggers(fc_estate, fc_resultRelInfo,
							 NULL, NULL,
							 fc_tid, NULL, fc_slot,
							 fc_recheckIndexes, NULL, false);

		list_free(fc_recheckIndexes);
	}
}

/*
 * 找到搜索槽元组并删除它，并执行任何约束
 * 和逐行触发器。
 *
 * 调用者有责任打开索引。
 */
void ExecSimpleRelationDelete(ResultRelInfo *fc_resultRelInfo,
						 EState *fc_estate, EPQState *fc_epqstate,
						 TupleTableSlot *fc_searchslot)
{
	bool		fc_skip_tuple = false;
	Relation	fc_rel = fc_resultRelInfo->ri_RelationDesc;
	ItemPointer fc_tid = &fc_searchslot->tts_tid;

	CheckCmdReplicaIdentity(fc_rel, CMD_DELETE);

	/* 行删除之前触发器 */
	if (fc_resultRelInfo->ri_TrigDesc &&
		fc_resultRelInfo->ri_TrigDesc->trig_delete_before_row)
	{
		fc_skip_tuple = !ExecBRDeleteTriggers(fc_estate, fc_epqstate, fc_resultRelInfo,
										   fc_tid, NULL, NULL);
	}

	if (!fc_skip_tuple)
	{
		/* 好的，删除元组 */
		simple_table_tuple_delete(fc_rel, fc_tid, fc_estate->es_snapshot);

		/* 行删除之后触发器 */
		ExecARDeleteTriggers(fc_estate, fc_resultRelInfo,
							 fc_tid, NULL, NULL, false);
	}
}

/*
 * 检查命令是否可以使用当前副本标识执行。
 */
void CheckCmdReplicaIdentity(Relation fc_rel, CmdType fc_cmd)
{
	PublicationDesc fc_pubdesc;

	/*
	 * 跳过对分区表的副本标识检查，因为操作实际上是在叶子分区上执行的。
	 */
	if (fc_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		return;

	/* 我们只需要对更新和删除进行检查。 */
	if (fc_cmd != CMD_UPDATE && fc_cmd != CMD_DELETE)
		return;

	/*
	 * 只有在所有引用于来自发布的行过滤器中
	 * 的列都是有效的情况下，才能安全地执行更新/删除
	 * - 例如，当所有引用的列都是副本标识的一部分
	 * 或该表不发布更新或删除时。
	 *
	 * XXX 我们可以通过首先检查这些发布是否有
	 * 针对此关系的行过滤器来优化它。如果没有且关系
	 * 有副本标识，则可以避免构建描述符，但由于
	 * 这只发生一次，似乎没有必要增加额外的
	 * 复杂性。
	 */
	RelationBuildPublicationDesc(fc_rel, &fc_pubdesc);
	if (fc_cmd == CMD_UPDATE && !fc_pubdesc.rf_valid_for_update)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
				 errmsg("cannot update table \"%s\"",
						RelationGetRelationName(fc_rel)),
				 errdetail("Column used in the publication WHERE expression is not part of the replica identity.")));
	else if (fc_cmd == CMD_UPDATE && !fc_pubdesc.cols_valid_for_update)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
				 errmsg("cannot update table \"%s\"",
						RelationGetRelationName(fc_rel)),
				 errdetail("Column list used by the publication does not cover the replica identity.")));
	else if (fc_cmd == CMD_DELETE && !fc_pubdesc.rf_valid_for_delete)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
				 errmsg("cannot delete from table \"%s\"",
						RelationGetRelationName(fc_rel)),
				 errdetail("Column used in the publication WHERE expression is not part of the replica identity.")));
	else if (fc_cmd == CMD_DELETE && !fc_pubdesc.cols_valid_for_delete)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
				 errmsg("cannot delete from table \"%s\"",
						RelationGetRelationName(fc_rel)),
				 errdetail("Column list used by the publication does not cover the replica identity.")));

	/* 如果关系具有副本标识，我们总是好的。 */
	if (OidIsValid(RelationGetReplicaIndex(fc_rel)))
		return;

	/* REPLICA IDENTITY FULL 也适用于 UPDATE/DELETE。 */
	if (fc_rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
		return;

	/*
	 * 这是 UPDATE/DELETE 并且没有复制标识。
	 *
	 * 检查表是否发布的是更新或删除。
	 */
	if (fc_cmd == CMD_UPDATE && fc_pubdesc.pubactions.pubupdate)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot update table \"%s\" because it does not have a replica identity and publishes updates",
						RelationGetRelationName(fc_rel)),
				 errhint("To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.")));
	else if (fc_cmd == CMD_DELETE && fc_pubdesc.pubactions.pubdelete)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot delete from table \"%s\" because it does not have a replica identity and publishes deletes",
						RelationGetRelationName(fc_rel)),
				 errhint("To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.")));
}


/*
 * 检查我们是否支持写入特定的 relkind。
 *
 * nspname 和 relname 仅用于错误报告。
 */
void CheckSubscriptionRelkind(char fc_relkind, const char *fc_nspname,
						 const char *fc_relname)
{
	if (fc_relkind != RELKIND_RELATION && fc_relkind != RELKIND_PARTITIONED_TABLE)
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("cannot use relation \"%s.%s\" as logical replication target",
						fc_nspname, fc_relname),
				 errdetail_relkind_not_supported(fc_relkind)));
}
